package com.hyperchain.config.mq.work;

import cn.hutool.core.util.StrUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Random;
import java.util.concurrent.*;

@Component
@Slf4j
public class HelloReceiver2 {

    private final Map<Integer, BlockingQueue<String>> queueMap = new ConcurrentHashMap<>();
    private final LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
    private final Random random = new Random();



    //乱序消费
//    @RabbitListener(queues = "q_hello")
//    public void listen(String object, Message message, Channel channel) {
//        executorService.submit(() -> processMessage(object, message, channel));
//    }
//
//    private void processMessage(String object, Message message, Channel channel) {
//        long deliveryTag = message.getMessageProperties().getDeliveryTag();
//        try {
//            // 模拟处理时间
//            int processingTime = random.nextInt(3000) + 500;
//            Thread.sleep(processingTime);
//            log.info("消费成功：{}, 消息内容: {}", deliveryTag, object);
//            channel.basicAck(deliveryTag, false);
//        } catch (Exception e) {
//            log.error("签收失败", e);
//            try {
//                channel.basicNack(deliveryTag, false, true);
//            } catch (Exception exception) {
//                log.error("拒签失败", exception);
//            }
//        }
//    }


    //有序消费
    @RabbitListener(queues = "q_hello")
    public void listen(Message message, Channel channel) {
        String messageBody = new String(message.getBody());
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 将消息放入队列以按顺序处理
        queue.offer(messageBody);
        processMessages(deliveryTag , channel);
        //executorService.submit(() -> processMessage(messageBody, deliveryTag, channel));
    }

    private void processMessages(long deliveryTag , Channel channel) {
        while (!queue.isEmpty()) {
            String messageBody = queue.poll(); // 获取并移除队首元素
            if (StrUtil.isNotEmpty(messageBody)) {
                processMessage(messageBody , deliveryTag , channel);
            }
        }
    }

    private void processMessage(String message, long deliveryTag, Channel channel) {
        try {
            // 模拟处理时间
            int processingTime = (int) (Math.random() * 3000) + 500;
            Thread.sleep(processingTime);
            log.info("消费成功：{}, 消息内容: {}", deliveryTag, message);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理消息失败，拒绝消息：{}", deliveryTag, e);
            try {
                channel.basicNack(deliveryTag, false, true);
            } catch (Exception ex) {
                log.error("拒签失败", ex);
            }
        }
    }
}
